package com.control;

import java.util.Map;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;

import com.dao.DemoDao;

import hyl.core.MyFun;
import hyl.core.run.IRun;
import hyl.core.run.MyRun;
import hyl.base.mq.kafka.MyKafka;

@Controller
public class DemoMqControl {

	MyKafka kfks = MyKafka.getInsConsumer("client1");
	private final String _topic = "demomq";
	@Autowired 
	private DemoDao demoDao;
	IRun DONE = new IRun() {
		@Override
		public void run(Object obj) {
			//  System.out.println(new String((byte[])obj));				
			ConsumerRecord<String, String> cr = (ConsumerRecord) obj;
		
			if (cr.key().equals("insUser")) {
				insUser(cr.value());
			} else if (cr.key().equals("list")) {
				demoDao.list(MyFun.json2Map(cr.value()));
			}
		}
	};

	public DemoMqControl() {
	
	}
	public void start() {
		MyRun mr=	MyRun.getInstance("测试",1);
		mr.execute(new Runnable() {
			@Override
			public void run() {
				kfks.receive(_topic, DONE);				
			}			
		});		
		System.out.println("接收消息的服务已经启动");
	}

	public void insUser(String text) {
		Map<String, String> obj = MyFun.json2Map(text);
	}
}
